package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.func.KeywordUDTF;
import com.atguigu.gmall.realtime.bean.KeywordBean;
import com.atguigu.gmall.realtime.common.GmallConstant;
import com.atguigu.gmall.realtime.util.MyClickhouseUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author Felix
 * @date 2022/10/6
 * 流量域：来源关键词聚合统计
 * 需要启动的进程
 * zk、kafka、flume、DwdTrafficBaseLogSplit、DwsTrafficSourceKeywordPageViewWindow
 */
public class DwsTrafficSourceKeywordPageViewWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //1.4 将自定义函数注册到表执行环境中
        tableEnv.createTemporarySystemFunction("ik_analyze", KeywordUDTF.class);

        //TODO 2.检查点相关的设置
        /*//2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置job取消之后检查点是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置两个检查点之间最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop202:8020/xxxx");
        //2.7 设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");*/

        //TODO 3.从kakfa的dwd_traffic_page_log主题中读取页面日志 创建动态表   执行Watermark以及提取事件时间字段
        tableEnv.executeSql("CREATE TABLE page_log (\n" +
            "    common map<string,string>,\n" +
            "    page map<string,string>, \n" +
            "    ts BIGINT,\n" +
            "    rowtime as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000)),\n" +
            "    WATERMARK FOR rowtime AS rowtime - INTERVAL '3' SECOND\n" +
            ") " + MyKafkaUtil.getKafkaDDL("dwd_traffic_page_log", "dws_traffic_keyword_group"));


        // tableEnv.executeSql("select * from page_log").print();

        //TODO 4.过滤出搜索行为
        Table searchTable = tableEnv.sqlQuery("select\n" +
            "    page['item'] fullword,\n" +
            "    rowtime\n" +
            "from page_log where page['last_page_id']='search' and page['item_type'] = 'keyword' and page['item'] is not null");
        tableEnv.createTemporaryView("search_table", searchTable);
        // tableEnv.executeSql("select * from search_table").print();

        //TODO 5.使用自定义函数进行分词  并将分词结果和原表字段进行关联
        Table splitTable = tableEnv.sqlQuery("SELECT keyword,rowtime\n" +
            "FROM search_table,LATERAL TABLE(ik_analyze(fullword)) t(keyword)");
        tableEnv.createTemporaryView("split_table", splitTable);
        // tableEnv.executeSql("select * from split_table").print();

        //TODO 6.分组、开窗、聚合计算
        Table resTable = tableEnv.sqlQuery("select \n" +
            "    DATE_FORMAT(TUMBLE_START(rowtime, INTERVAL '10' SECOND), 'yyyy-MM-dd HH:mm:ss') stt,\n" +
            "    DATE_FORMAT(TUMBLE_END(rowtime, INTERVAL '10' SECOND), 'yyyy-MM-dd HH:mm:ss')  edt,\n" +
            "    '" + GmallConstant.KEYWORD_SEARCH + "' source,\n" +
            "    keyword, \n" +
            "    count(*) keyword_count,\n" +
            "    UNIX_TIMESTAMP()*1000 ts\n" +
            "from split_table group by keyword,TUMBLE(rowtime, INTERVAL '10' SECOND)");
        // tableEnv.createTemporaryView("res_table", resTable);
        // tableEnv.executeSql("select * from res_table").print();

        //TODO 7.将动态表中转换为流
        DataStream<KeywordBean> keywordDS = tableEnv.toAppendStream(resTable, KeywordBean.class);

        //TODO 8.将流中的数据写到Clickhouse表中
        keywordDS.print(">>>>");
        keywordDS.addSink(MyClickhouseUtil.<KeywordBean>getSinkFunction("insert into dws_traffic_source_keyword_page_view_window values(?,?,?,?,?,?)"));

        env.execute();
    }
}
